package com.abyss.window;


import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * 演示窗口对象的apply方法
 * 做一个word count
 */
public class WindowApplyDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // Source, 接收单词, 单词按照空格分隔
        DataStreamSource<String> socketTextStream = env.socketTextStream("node1", 9999);

        // 将单词 转变为元组
        SingleOutputStreamOperator<Tuple2<String, Integer>> wordWithOne = socketTextStream.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception {
                String[] words = value.split(" ");
                for (String word : words) {
                    out.collect(Tuple2.of(word, 1));
                }
            }
        });

        // 加窗口, 按单词分组, 窗口时间10秒
        WindowedStream<Tuple2<String, Integer>, Tuple, TimeWindow> windowedStream = wordWithOne
                .keyBy(0).window(TumblingProcessingTimeWindows.of(Time.seconds(10)));

        // 应用apply方法, 对单词个数进行统计
        SingleOutputStreamOperator<Tuple2<String, Integer>> apply = windowedStream.apply(new WindowFunction<Tuple2<String, Integer>, Tuple2<String, Integer>, Tuple, TimeWindow>() {
            @Override
            public void apply(Tuple tuple, TimeWindow window, Iterable<Tuple2<String, Integer>> input, Collector<Tuple2<String, Integer>> out) throws Exception {
                int sum = 0;
                String key = null;
                for (Tuple2<String, Integer> wordWithOne : input) {
                    sum += wordWithOne.f1;
                    key = wordWithOne.f0;
                }
                out.collect(Tuple2.of(key, sum));
            }
        });

        apply.print();

        env.execute();
    }
}